package com.sun.rabbitMQ.Confirms;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.sun.rabbitMQ.utils.FactoryUtils;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

public class ProducerACK {
    public static void main(String[] args) throws InterruptedException, TimeoutException, IOException {
//        publishMessageOne();
//        publishMessageMore();
        publishMessageAsy();
    }
    //单个确认  发送100 条  37毫秒
    public static void publishMessageOne() throws IOException, TimeoutException, InterruptedException {
        Channel channel = FactoryUtils.getChannel();
        String uuid = UUID.randomUUID().toString();
        channel.queueDeclare(uuid,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long beginTime = System.currentTimeMillis();
        for (int i = 0; i < 100; i++) {
            String message = i + "";
            channel.basicPublish("",uuid,null,message.getBytes());
            //单个消息发布确认
            boolean flag = channel.waitForConfirms();
            if (flag){
                System.out.println("消息发送成功");
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println(endTime - beginTime);
    }
    //批量100条 20毫秒
    public static void publishMessageMore() throws IOException, TimeoutException, InterruptedException{
        Channel channel = FactoryUtils.getChannel();
        String uuid = UUID.randomUUID().toString();

        channel.queueDeclare(uuid,false,false,false,null);
        channel.confirmSelect();
        long beginTime = System.currentTimeMillis();
        for (int i = 0; i < 100; i++) {
            String message = i + "";
            channel.basicPublish("",uuid,null,message.getBytes("utf-8"));
            //批量消息发布确认
            if (i%10 == 0){
                boolean flag = channel.waitForConfirms();
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println(endTime - beginTime);
    }
    //异步确认发布  18毫秒
    public static void publishMessageAsy() throws IOException, TimeoutException, InterruptedException{
        ConcurrentSkipListMap<Long,String> concurrentSkipListMap = new ConcurrentSkipListMap();
        Channel channel = FactoryUtils.getChannel();
        String uuid = UUID.randomUUID().toString();
        channel.queueDeclare(uuid,false,false,false,null);
        channel.confirmSelect();
        long beginTime = System.currentTimeMillis();
        //准备消息得监听器  监听那些消息成功了，那些消息失败了
        //监听成功
        ConfirmCallback ackCallback = (deliveryTag,multiple)->{
            if (multiple){
                //批量确认  将处理成功的批量删除
                ConcurrentNavigableMap<Long, String> longStringConcurrentNavigableMap = concurrentSkipListMap.headMap(deliveryTag);
                longStringConcurrentNavigableMap.clear();
            }else {
                concurrentSkipListMap.remove(deliveryTag);
            }

            System.out.println("确认消息 " + deliveryTag);
        };
        //监听失败   deliveryTag  消息得标识      multiple 是否为批量确认
        ConfirmCallback nackCallback = (deliveryTag,multiple)->{
            System.out.println("未确认消息 " + deliveryTag);
        };
        //异步通知
        channel.addConfirmListener(ackCallback,nackCallback);
        for (int i = 0; i < 100; i++) {
            String message = "异步 " + i;
            concurrentSkipListMap.put(channel.getNextPublishSeqNo(),message);
            channel.basicPublish("",uuid,null,message.getBytes("utf-8"));
        }
        long endTime = System.currentTimeMillis();
        System.out.println(endTime - beginTime);
    }
}
